package com.fuda.trade

import com.fuda.trade.ZenohManager.ZenohCallback
import io.zenoh.Session
import io.zenoh.keyexpr.KeyExpr
import io.zenoh.query.Reply
import io.zenoh.sample.Sample
import io.zenoh.selector.Selector
import kotlinx.coroutines.*
import org.junit.jupiter.api.AfterAll
import org.junit.jupiter.api.BeforeAll
import org.junit.jupiter.api.Test
import java.time.LocalDateTime
import java.util.Optional
import java.util.concurrent.*

class ZSubTest {


    @Test
    fun testSubscribe()  = runBlocking<Unit>  {
        val latch = CountDownLatch(1)
        println("Press CTRL-C to quit...")
        val mainScope = CoroutineScope(Dispatchers.IO)

        val subC =  mainScope.launch {
            Session.open().use { session->
                KeyExpr.tryFrom(ZenohManager.TRADE_ORDERS_TOPIC).use { keyExpr ->
                    println("Declaring Subscriber on '$keyExpr'...")
                    session.declareSubscriber(keyExpr).res().use { subscriber ->
                        val receiver: BlockingQueue<Optional<Sample>> =
                            checkNotNull(subscriber.receiver)
                        while (true) {
                            val now = LocalDateTime.now()
                            val wrapper = receiver.take()
                            val sample = wrapper.get()
                            println(">> $now [Subscriber] Received ${sample.kind} ('${sample.keyExpr}': '${sample.value}')")
                        }
                    }
                }
            }
        }

        val rc =  mainScope.launch {
            Session.open().use { session->
                KeyExpr.tryFrom(ZenohManager.ZENOH_RESPONSE_TOPIC).use { keyExpr ->
                    println("Declaring Subscriber on '$keyExpr'...")
                    session.declareSubscriber(keyExpr).res().use { subscriber ->
                        val receiver: BlockingQueue<Optional<Sample>> =
                            checkNotNull(subscriber.receiver)
                        while (true) {
                            val now = LocalDateTime.now()
                            val wrapper = receiver.take()
                            val sample = wrapper.get()
                            println(">> $now [Subscriber] Received ${sample.kind} ('${sample.keyExpr}': '${sample.value}')")
                        }
                    }
                }
            }
        }

        val hbc =  mainScope.launch {
            Session.open().use { session->
                KeyExpr.tryFrom(ZenohManager.ZENOH_HEARTBEAT_TOPIC).use { keyExpr ->
                    println("Declaring Subscriber on '$keyExpr'...")
                    session.declareSubscriber(keyExpr).res().use { subscriber ->
                        val receiver: BlockingQueue<Optional<Sample>> =
                            checkNotNull(subscriber.receiver)
                        while (true) {
                            val now = LocalDateTime.now()
                            val wrapper = receiver.take()
                            val sample = wrapper.get()
                            println(">> $now [Subscriber] Received ${sample.kind} ('${sample.keyExpr}': '${sample.value}')")
                        }
                    }
                }
            }
        }

        val getC =  mainScope.launch {
            Session.open().use { session ->
                var count = 0
                while(true){
                    ZenohManager.strategiesSelector?.let {
                        select(session, it, object : ZenohManager.ZenohCallback {
                            override fun onSuccess(sample: Sample) {
                                // Process the successful sample here
//                                println("[Selector] received: ${sample.keyExpr} - ${sample.value}")
                            }

                            override fun onError(error: String) {
                                // Handle errors here
//                                println("[Selector] Error occurred: $error")
                            }
                        })
                    }
                    ZenohManager.ordersSelector?.let {
                        select(session, it, object : ZenohManager.ZenohCallback {
                            override fun onSuccess(sample: Sample) {
                                // Process the successful sample here
//                                println("[Selector] received: ${sample.keyExpr} - ${sample.value}")
                            }

                            override fun onError(error: String) {
                                // Handle errors here
//                                println("[Selector] Error occurred: $error")
                            }
                        })
                    }
                    ZenohManager.holdingsSelector?.let {
                        select(session, it, object : ZenohManager.ZenohCallback {
                            override fun onSuccess(sample: Sample) {
                                // Process the successful sample here
//                                println("[Selector] received: ${sample.keyExpr} - ${sample.value}")
                            }

                            override fun onError(error: String) {
                                // Handle errors here
//                                println("[Selector] Error occurred: $error")
                            }
                        })
                    }

                    count++
                    TimeUnit.SECONDS.sleep(30) // next round get
                }
            }
        }


        // Wait until the test is interrupted
        latch.await()
        mainScope.cancel()

        // Await the coroutines to ensure they finish
        withContext(Dispatchers.IO) {
            subC.join()
            rc.join()
            hbc.join()
            getC.join()
        }
    }

    private fun select(session:Session, selector: Selector, callback: ZenohCallback){
        val now = LocalDateTime.now()
        println( ">> $now (select) Performing on '$selector'...")

        val receiver: BlockingQueue<Optional<Reply>> = checkNotNull(session.get(selector).res())

        while (true) {
            val now = LocalDateTime.now()
            val wrapper = receiver.take()

            if (wrapper == null) {
                println( ">> $now (select)  break null")
                break
            }

            val reply = wrapper.orElse(null)

            if (reply == null) {
                println(">> $now (select) break null")
                break
            }

            when (reply) {
                is Reply.Success -> {
                    val successReply = reply as Reply.Success
                    println(">>  $now (select)  received: ${successReply.sample.keyExpr} - ${successReply.sample.value}")
                    callback.onSuccess(successReply.sample)
                }
                is Reply.Error -> {
                    val errorReply = reply as Reply.Error
                    println(">> $now (select) Received (ERROR: '" + errorReply.error + "')")
                    callback.onError(errorReply.error.toString())
                }
            }
        }
    }

    companion object {

        @JvmStatic
        @AfterAll
        fun tearDown(): Unit {
        }

        @JvmStatic
        @BeforeAll
        fun setUp(): Unit {
            println("Current DYLD_LIBRARY_PATH: ${System.getenv("DYLD_LIBRARY_PATH")}")
        }
    }
}

